//package com.zhz.mall.portal.kakfa;
//
//import cn.hutool.core.collection.CollUtil;
//import cn.hutool.core.util.ArrayUtil;
//import com.zhz.mall.common.util.EnvironmentUtils;
//import com.zhz.mall.common.util.ExceptionUtils;
//import com.zhz.mall.common.util.ccollect.CollStreamUtil;
//import com.zhz.mall.common.util.listener.ApplicationStartedListener;
//import com.zhz.mall.common.util.threadpool.ThreadPoolUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//import org.springframework.context.ConfigurableApplicationContext;
//import org.springframework.stereotype.Component;
//import org.springframework.util.LinkedMultiValueMap;
//import org.springframework.util.MultiValueMap;
//
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//import java.util.Objects;
//import java.util.concurrent.Future;
//import java.util.concurrent.FutureTask;
//import java.util.concurrent.TimeUnit;
//
///**
// * KafkaDispatcher
// * <p>
// * kafka 消息分发器, 统一监听 kafka 消息, 并进行消息分发处理
// *
// * @author zhangjiawei@sanyygp.com
// * @version 1.0
// * @date 2021/11/18 14:50
// */
//@Slf4j
//@Component
//public class KafkaDispatcher implements ApplicationStartedListener {
//
//    private final MultiValueMap<String, KafkaMessageListener> listeners = new LinkedMultiValueMap<>();
//    private final Map<String, KafkaConsumer> kafkaConsumers = new HashMap<>();
//
//
//    /**
//     * 添加消息监听器
//     */
//    void addListener(KafkaMessageListener listener) {
//        listeners.add(listener.getTopic(), listener);
//    }
//
//
//    /**
//     * 消息分发处理
//     */
//    void doDispatch(ConsumerRecord record) {
//        String topic = record.topic();
//        String value = record.value().toString();
//        log.info("接收 kafka 消息, topic[{}], message: {}", topic, value);
//
//        List<KafkaMessageListener> targetListeners = this.listeners.get(topic);
//        if (CollUtil.isEmpty(targetListeners)) {
//            log.error("no listener found for topic[{}], message: {}", topic, value);
//            return;
//        }
//
//        long timeoutMillis = EnvironmentUtils.getProperty("spring.kafka.consumer.task-timeout", long.class, 60_000L);
//
//        // 并发执行（当有多个消费者时）
//        Map<KafkaMessageListener, Future<?>> taskHolder = new HashMap<>();
//        for (int i = 0, size = targetListeners.size(); i < size; i++) {
//            KafkaMessageListener listener = targetListeners.get(i);
//            String listenerName = listener.getName();
//
//            Runnable task = () -> {
//                long startTime = System.currentTimeMillis();
//
//                try {
//                    listener.onMessage(record);
//                } catch (Throwable cause) {
//                    log.error("kafka 消费者[{}]处理异常: {}, topic[{}], message: {}", listenerName,
//                            cause.getMessage(), topic, value, cause);
//                }
//
//                long endTime = System.currentTimeMillis();
//                long costTimeMills = endTime - startTime;
//                boolean isTimeout = costTimeMills > timeoutMillis;
//                if (isTimeout) {
//                    log.error("kafka 消费者[{}]执行时间异常, 执行耗时[{}], 请尽快检查业务逻辑, topic[{}], message: {}",
//                            listenerName, costTimeMills, topic, value);
//                }
//
//            };
//
//            boolean async = size > 1 && i != size - 1;
//            Future<?> taskFuture;
//            if (async) {
//                // 前 n - 1 个交给线程池
//                taskFuture = ThreadPoolUtils.submit(task);
//            } else {
//                // 最后 1 个自己执行
//                // 注意: 如果业务代码执行时间太长可能会造成 consumer rebalance, 所以一定合理处理业务逻辑
//                FutureTask<Void> futureTask = new FutureTask<>(task, null);
//                futureTask.run();
//                taskFuture = futureTask;
//            }
//
//            taskHolder.put(listener, taskFuture);
//        }
//
//
//        // 等待任务执行完成
//        long remainTimeoutMillis = timeoutMillis;
//        for (Map.Entry<KafkaMessageListener, Future<?>> entry : taskHolder.entrySet()) {
//            KafkaMessageListener listener = entry.getKey();
//            Future<?> taskFuture = entry.getValue();
//
//            long startTime = System.currentTimeMillis();
//            Throwable cause = ThreadPoolUtils.safeWait(taskFuture, remainTimeoutMillis, TimeUnit.MILLISECONDS, true);
//            long endTime = System.currentTimeMillis();
//            long costTimeMills = endTime - startTime;
//            remainTimeoutMillis -= costTimeMills;
//
//            if (Objects.nonNull(cause)) {
//                log.error("kafka 消费者[{}]处理异常: {}, topic[{}], message: {}", listener.getName(),
//                        ExceptionUtils.getMessage(cause), topic, value, cause);
//            }
//
//            if (remainTimeoutMillis <= 0) {
//                break;
//            }
//        }
//
//    }
//
//
//    /**
//     * 获取监听的 topic
//     */
//    String[] getTopics() {
//        return CollStreamUtil.toArray(listeners.keySet());
//    }
//
//
//
//
//    @Override
//    public void onApplicationStarted(ConfigurableApplicationContext context) {
//        String configKey = "spring.kafka.consumer.auto-startup";
//        boolean autoStartup = context.getEnvironment().getProperty(configKey, boolean.class, true);
//        if (autoStartup) {
//            String[] topics = getTopics();
//            if (CollUtil.isEmpty(kafkaConsumers) && ArrayUtil.isNotEmpty(topics)) {
//                // init kafka consumer
//                for (String topic : topics) {
//                    KafkaConsumer kafkaConsumer = context.getBean(KafkaConsumer.class, topic);
//                    kafkaConsumers.put(topic, kafkaConsumer);
//                }
//            }
//        } else {
//            log.info("kafka consumer not start, config key: {}", configKey);
//        }
//    }
//
//
//
//}
